package com.study.zookeeper.impl;

import com.study.common.ZookeeperConfig;
import com.study.zookeeper.IZookeeperRegister;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by zhangbo on 2020/5/30.
 */
@Slf4j
@Component
public class ZookeeperRegisterImpl implements IZookeeperRegister, CommandLineRunner {

    /**
     * zookeeper配置
     */
    @Autowired
    private ZookeeperConfig zookeeperConfig;


    private volatile ZooKeeper zooKeeper;


    public static final Map<String, byte[]> REGISTER_INFO_OBJECT = new ConcurrentHashMap<>(32);

    @Override
    public boolean saveRegister(String path, byte[] registerInfo) throws IOException, KeeperException, InterruptedException {
        ZooKeeper zooKeeper = getZooKeeper();
        String registerPath = getRegisterPath(path);
        /*创建注册目录，并设置注册数据*/
        loopCreatePath(zooKeeper, registerPath, true);
        Stat stat = zooKeeper.setData(registerPath, registerInfo, -1);
        REGISTER_INFO_OBJECT.put(path, registerInfo);
        return true;
    }

    private String getRegisterPath(String path) {
        String registerPath = zookeeperConfig.getZookeeperRegisterPath();
        if (!registerPath.endsWith("/") && !path.startsWith("/")) {
            registerPath = registerPath + "/" + path;
        }
        return registerPath;
    }

    @Override
    public byte[] getRegister(String path) throws IOException, KeeperException, InterruptedException {
        ZooKeeper zooKeeper = getZooKeeper();
        String registerPath = getRegisterPath(path);
        Stat exists = zooKeeper.exists(registerPath, false);
        if (null == exists) {
            return null;
        }
        byte[] data = zooKeeper.getData(registerPath, false, null);
        return data;
    }

    @Override
    public List<String> getRegisterPathList(String path) throws IOException, KeeperException, InterruptedException {
        ZooKeeper zooKeeper = getZooKeeper();
        String registerPath = getRegisterPath(path);
        List<String> children = zooKeeper.getChildren(registerPath, false);
        return children;
    }

    @Override
    public boolean deleteRegisterPath(String path) throws IOException, KeeperException, InterruptedException {
        ZooKeeper zooKeeper = getZooKeeper();
        String registerPath = getRegisterPath(path);
        Stat exists = zooKeeper.exists(registerPath, false);
        if (null == exists) {
            return false;
        }
        zooKeeper.delete(registerPath, -1);
        return true;
    }


    /**
     * 获取zookeeper客户端
     *
     * @return
     * @throws IOException
     */
    private ZooKeeper getZooKeeper() throws IOException {
        if (zooKeeper == null || !zooKeeper.getState().isAlive()) {
            synchronized (this) {
                if (zooKeeper == null || !zooKeeper.getState().isAlive()) {
                    zooKeeper = new ZooKeeper(zookeeperConfig.getZkAddress(), zookeeperConfig.getTimeout(), new Watcher() {
                        @Override
                        public void process(WatchedEvent watchedEvent) {
                            /*如果连接断开、或者关闭、或者过期，则需要重新构建连接*/
                            if (Event.KeeperState.Disconnected.equals(watchedEvent.getState()) ||
                                    Event.KeeperState.Closed.equals(watchedEvent.getState()) ||
                                    Event.KeeperState.Expired.equals(watchedEvent.getState())) {
                                log.warn("event state:{}", watchedEvent.getState());
                                log.info("重新连接");
                                syncInitServiceInfo();
                            }

                        }
                    });
                }
                if (!isAlive()) {
                    getZooKeeper();
                }
            }
        }
        return zooKeeper;
    }

    /**
     * 服务重连注册
     */
    private void syncInitServiceInfo() {
        log.info("开始服务重连注册操作");
        for (String key : REGISTER_INFO_OBJECT.keySet()) {
            try {
                log.info("开始注册接口：{}", key);
                saveRegister(key, REGISTER_INFO_OBJECT.get(key));
            } catch (IOException e) {
                log.error("服务注册异常：{}", key);
            } catch (KeeperException e) {
                log.error("服务注册异常：{}", key);
            } catch (InterruptedException e) {
                log.error("服务注册异常：{}", key);
            }
        }
        log.info("服务重连注册完成");
    }

    private boolean isAlive() {
        for (int i = 0; i < 3; i++) {
            if (!zooKeeper.getState().isAlive()) {
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                continue;
            }
            return true;
        }
        log.warn("更新连接状态失败");
        return false;
    }

    /**
     * 循环校验并创建父节点
     *
     * @param zooKeeper
     * @param path      创建目录
     * @param flag      最后节点是否创建为临时节点
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void loopCreatePath(ZooKeeper zooKeeper, String path, boolean flag) throws KeeperException, InterruptedException {
        if (StringUtils.isNotBlank(path)) {
            String[] paths = path.split("/");
            StringBuilder stringBuilder = new StringBuilder();
            int count = 0;
            for (String tmpPath : paths) {
                count++;
                if (StringUtils.isBlank(tmpPath)) {
                    continue;
                }
                stringBuilder.append("/").append(tmpPath);
                Stat exists = zooKeeper.exists(stringBuilder.toString(), false);
                if (null == exists) {
                    if (count == paths.length && flag) {
                        zooKeeper.create(stringBuilder.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                        log.info("创建zookeeper临时目录：{}成功", stringBuilder.toString());
                    } else {
                        zooKeeper.create(stringBuilder.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                        log.info("创建zookeeper目录：{}成功", stringBuilder.toString());
                    }
                }
            }
        }
    }

    private void close(ZooKeeper zooKeeper) {
        if (null != zooKeeper) {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                log.error("关闭zooKeeper客户端异常：{}", e);
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public void run(String... strings) throws Exception {
        getZooKeeper();
        log.info("实例化zookeeper客户端");
        TimeUnit.MILLISECONDS.sleep(100);
        log.info("开始创建实例父节点");
        loopCreatePath(zooKeeper, zookeeperConfig.getZookeeperRegisterPath(), false);
    }
}
